package mammuthus.proxy

import java.io.File
import java.util
import java.util.Date
import java.util.concurrent.{TimeUnit, Executors}
import com.google.inject.Inject
import com.webdm.common.io.FileUtil
import com.webdm.common.shell.CmdHealper
import mammuthus.client.{ConfClient, ZKClientWrapper, ConfCallBack}
import net.csdn.common.settings.Settings
import org.slf4j.LoggerFactory
import java.util.concurrent.atomic.AtomicBoolean

import webdm.letv.com.conf.util.ZKDistributedLock


/**
 * Created by fbj on 15-2-6 上午10:44.
 */
class MammuthusConfControl @Inject()(settings: Settings, confClient: ConfClient,zkClientWrapper:ZKClientWrapper) {

  val dir = settings.get("conf.localDir")
  val logger = LoggerFactory.getLogger(classOf[MammuthusConfControl])
  var hostname = "error"
  val outputDir = settings.get("conf.outputDir", dir)

  val confCallBack = new ConfCallBack {
    override def confChange(path: String, conf: String): Unit = {


      val p = (outputDir + "/" + path).replaceAll("/+", "/")
      logger.info("生成配置文件{}", p)
      if (conf == null) {
        FileUtil.writeFile(p, "")
      } else {
        FileUtil.writeFile(p, conf)
      }

    }

    override def confDelete(path: String): Unit = {

      val p = (outputDir + "/" + path).replaceAll("/+", "/")
      val f = new File(p)
      if (f.exists()) {

        f.delete()
        logger.info("生成配置文件{}", p)

      }
    }
  }

  val confNames = new util.HashSet[String]()

  /**
   * 检测需要同步的目录
   */
  def checkDir() = {

    new File(dir).list().foreach(p => {

      if (confNames.add(p)) {

        if(confClient.ifExist(p)){
          logger.info("添加监控目录{}", p)
          confClient.addDirListener("/" + p, confCallBack, hostname)
        }else{
            logger.info("zk目录不存在{}", p)
            val absolutePath = dir + "/" + p
            val localConfDir = new File(absolutePath)
            if(localConfDir.list().size > 0){
              val lock = new ZKDistributedLock("zk_file_lock_" + p,confClient.getZKClientWrapper.client)

              if(lock.lock()){

                  try{
                    if(!confClient.ifExist(p)) {
                      new File(absolutePath).list().foreach(subFile => {
                        val confFile = new File(absolutePath + "/" + subFile)
                        if (confFile.exists() && confFile.isFile) {
                          logger.info("添加zk文件{} -> {}", Array(confFile.getAbsoluteFile, p + "/" + subFile))
                          confClient.addConf(p + "/" + subFile, FileUtil.readFile(confFile.getAbsolutePath))
                        }
                      })
                    }
                  }catch {
                    case e: Exception => e.printStackTrace()
                  }

                lock.unLock()

              }

            }else{

              logger.info("本地目录没有内容{}", absolutePath)
            }

            //下次再检测,以便注册监听
            confNames.remove(p)

        }

      }

    })

  }

  val pools = Executors.newScheduledThreadPool(2)

  val status = new AtomicBoolean(false)

  def start(servers:String): Unit = {

    if (status.get()) {
      logger.info("proxy {} 已经启动过", hostname)
      return
    }

    val zkConf = settings.get("conf.zk", servers)
    logger.info("zkConf {}", zkConf)
    zkClientWrapper.configZkClient(zkConf)
    hostname = CmdHealper.exeAndReturn("hostname".split(" "));

    logger.info("启动proxy {}", hostname)
    val proxyDir = confClient.getConfRoot + "/_proxy/" + hostname
    val zkClient = confClient.getZKClientWrapper

    if (zkClient.ifExist(proxyDir)) {
      zkClient.removePath(proxyDir)
    }
    zkClient.addEphemeralPath(proxyDir, new Date().toLocaleString)

    pools.scheduleAtFixedRate(new Runnable() {
      override def run = {
        checkDir()
      }
    }, 0, 10, TimeUnit.SECONDS);
    status.set(true)
  }

}
